// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package segments

import (
	"context"
	"fmt"
	"sync"

	"go.uber.org/atomic"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
	"github.com/milvus-io/milvus/internal/querynodev2/segments/metricsutil"
	"github.com/milvus-io/milvus/pkg/v2/log"
	"github.com/milvus-io/milvus/pkg/v2/metrics"
	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v2/util/timerecord"
)

// searchSegments 在指定的段列表上执行搜索操作
// 调用此函数前所有段ID都已通过验证
func searchSegments(ctx context.Context, mgr *Manager, segments []Segment, segType SegmentType, searchReq *SearchRequest) ([]*SearchResult, error) {
	// 根据段类型设置搜索标签，用于指标统计
	searchLabel := metrics.SealedSegmentLabel  // 默认为已封段标签
	if segType == commonpb.SegmentState_Growing {
		searchLabel = metrics.GrowingSegmentLabel  // 如果是增长段，则使用增长段标签
	}

	// 创建结果通道，用于收集各段的搜索结果
	resultCh := make(chan *SearchResult, len(segments))
	
	// 定义搜索器函数，用于在单个段上执行搜索
	searcher := func(ctx context.Context, s Segment) error {
		// 记录搜索时间
		tr := timerecord.NewTimeRecorder("searchOnSegments")
		// 在段上执行实际搜索操作
		searchResult, err := s.Search(ctx, searchReq)
		if err != nil {
			return err
		}
		// 将搜索结果发送到结果通道
		resultCh <- searchResult
		
		// 更新指标统计
		elapsed := tr.ElapseSpan().Milliseconds()  // 计算耗时
		// 记录段搜索延迟指标
		metrics.QueryNodeSQSegmentLatency.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()),
			metrics.SearchLabel, searchLabel).Observe(float64(elapsed))
		// 记录每个向量的搜索延迟指标
		metrics.QueryNodeSegmentSearchLatencyPerVector.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()),
			metrics.SearchLabel, searchLabel).Observe(float64(elapsed) / float64(searchReq.GetNumOfQuery()))
		return nil
	}

	// 使用goroutines并发执行段搜索
	errGroup, ctx := errgroup.WithContext(ctx)
	// 收集没有索引的段ID，用于日志记录
	segmentsWithoutIndex := make([]int64, 0)
	
	// 遍历所有段，为每个段启动一个goroutine进行搜索
	for _, segment := range segments {
		seg := segment
		// 检查段是否存在指定搜索字段的索引
		if !seg.ExistIndex(searchReq.SearchFieldID()) {
			segmentsWithoutIndex = append(segmentsWithoutIndex, seg.ID())
		}
		
		// 启动goroutine执行搜索
		errGroup.Go(func() error {
			// 检查上下文是否已取消
			if ctx.Err() != nil {
				return ctx.Err()
			}

			var err error
			// 创建搜索段访问记录，用于指标统计
			accessRecord := metricsutil.NewSearchSegmentAccessRecord(getSegmentMetricLabel(seg))
			// 函数退出时完成访问记录
			defer func() {
				accessRecord.Finish(err)
			}()

			// 如果段是懒加载的，需要从磁盘缓存中加载
			if seg.IsLazyLoad() {
				// 设置懒加载超时上下文
				ctx, cancel := withLazyLoadTimeoutContext(ctx)
				defer cancel()

				var missing bool
				// 从磁盘缓存中执行搜索
				missing, err = mgr.DiskCache.Do(ctx, seg.ID(), searcher)
				if missing {
					// 如果缓存缺失，记录缓存缺失事件
					accessRecord.CacheMissing()
				}
				if err != nil {
					// 记录磁盘缓存搜索失败日志
					log.Warn("failed to do search for disk cache", zap.Int64("segID", seg.ID()), zap.Error(err))
				}
				return err
			}
			// 直接在段上执行搜索
			return searcher(ctx, seg)
		})
	}
	
	// 等待所有goroutine完成
	err := errGroup.Wait()
	// 关闭结果通道
	close(resultCh)

	// 收集所有搜索结果
	searchResults := make([]*SearchResult, 0, len(segments))
	for result := range resultCh {
		searchResults = append(searchResults, result)
	}

	// 如果有错误发生，清理搜索结果并返回错误
	if err != nil {
		DeleteSearchResults(searchResults)
		return nil, err
	}

	// 如果存在没有索引的段，记录调试日志
	if len(segmentsWithoutIndex) > 0 {
		log.Ctx(ctx).Debug("search growing/sealed segments without indexes", zap.Int64s("segmentIDs", segmentsWithoutIndex))
	}

	// 返回搜索结果
	return searchResults, nil
}
// searchSegmentsStreamly performs search on listed segments in a stream mode instead of a batch mode
// all segment ids are validated before calling this function
func searchSegmentsStreamly(ctx context.Context,
	mgr *Manager,
	segments []Segment,
	searchReq *SearchRequest,
	streamReduce func(result *SearchResult) error,
) error {
	searchLabel := metrics.SealedSegmentLabel
	searchResultsToClear := make([]*SearchResult, 0)
	var reduceMutex sync.Mutex
	var sumReduceDuration atomic.Duration
	searcher := func(ctx context.Context, seg Segment) error {
		// record search time
		tr := timerecord.NewTimeRecorder("searchOnSegments")
		searchResult, searchErr := seg.Search(ctx, searchReq)
		searchDuration := tr.RecordSpan().Milliseconds()
		if searchErr != nil {
			return searchErr
		}
		reduceMutex.Lock()
		searchResultsToClear = append(searchResultsToClear, searchResult)
		reducedErr := streamReduce(searchResult)
		reduceMutex.Unlock()
		reduceDuration := tr.RecordSpan()
		if reducedErr != nil {
			return reducedErr
		}
		sumReduceDuration.Add(reduceDuration)
		// update metrics
		metrics.QueryNodeSQSegmentLatency.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()),
			metrics.SearchLabel, searchLabel).Observe(float64(searchDuration))
		metrics.QueryNodeSegmentSearchLatencyPerVector.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()),
			metrics.SearchLabel, searchLabel).Observe(float64(searchDuration) / float64(searchReq.GetNumOfQuery()))
		return nil
	}

	// calling segment search in goroutines
	errGroup, ctx := errgroup.WithContext(ctx)
	log := log.Ctx(ctx)
	for _, segment := range segments {
		seg := segment
		errGroup.Go(func() error {
			if ctx.Err() != nil {
				return ctx.Err()
			}

			var err error
			accessRecord := metricsutil.NewSearchSegmentAccessRecord(getSegmentMetricLabel(seg))
			defer func() {
				accessRecord.Finish(err)
			}()
			if seg.IsLazyLoad() {
				log.Debug("before doing stream search in DiskCache", zap.Int64("segID", seg.ID()))
				ctx, cancel := withLazyLoadTimeoutContext(ctx)
				defer cancel()

				var missing bool
				missing, err = mgr.DiskCache.Do(ctx, seg.ID(), searcher)
				if missing {
					accessRecord.CacheMissing()
				}
				if err != nil {
					log.Warn("failed to do search for disk cache", zap.Int64("segID", seg.ID()), zap.Error(err))
				}
				log.Debug("after doing stream search in DiskCache", zap.Int64("segID", seg.ID()), zap.Error(err))
				return err
			}
			return searcher(ctx, seg)
		})
	}
	err := errGroup.Wait()
	DeleteSearchResults(searchResultsToClear)
	if err != nil {
		return err
	}
	metrics.QueryNodeReduceLatency.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()),
		metrics.SearchLabel,
		metrics.ReduceSegments,
		metrics.StreamReduce).Observe(float64(sumReduceDuration.Load().Milliseconds()))
	log.Debug("stream reduce sum duration:", zap.Duration("duration", sumReduceDuration.Load()))
	return nil
}
// SearchHistorical 在历史段上执行搜索操作
// 如果未指定segIDs，将搜索partIDs指定的所有历史段
// 如果指定了segIDs，将只在segIDs指定的段上搜索
// 如果partIDs为空，表示搜索已加载集合的所有分区或所有已加载的分区
/* 各参数详细说明：
ctx context.Context - 上下文参数
用于控制请求的生命周期
可以用来取消操作或设置超时
传递请求相关的元数据

*manager Manager - 段管理器
负责管理所有段的生命周期
提供段的获取、验证、缓存等操作
用于访问和操作目标段

*searchReq SearchRequest - 搜索请求
包含具体的搜索参数和条件
包括查询向量、搜索字段、topk值、度量类型等
定义了要执行的搜索操作的具体内容

collID int64 - 集合ID
指定要搜索的集合标识符
用于定位特定的集合数据

partIDs []int64 - 分区ID列表
指定要搜索的分区列表
如果为空，则表示搜索集合中的所有分区
用于限定搜索范围到特定分区

segIDs []int64 - 段ID列表
指定要搜索的具体段ID列表
如果为空，则搜索指定分区中的所有历史段
用于精确控制搜索的段范围

返回值：
*[]SearchResult - 搜索结果列表，包含每个段的搜索结果
[]Segment - 实际被搜索的段列表
error - 可能的错误信息，如果搜索成功则为nil
 */
func SearchHistorical(ctx context.Context, manager *Manager, searchReq *SearchRequest, collID int64, partIDs []int64, segIDs []int64) ([]*SearchResult, []Segment, error) {
	// 检查上下文是否已取消
	if ctx.Err() != nil {
		return nil, nil, ctx.Err()
	}

	// 验证历史段的有效性
	segments, err := validateOnHistorical(ctx, manager, collID, partIDs, segIDs)
	if err != nil {
		return nil, nil, err
	}
	
	// 在验证通过的段上执行搜索操作
	// 使用SegmentTypeSealed类型表示历史段（已封段）
	searchResults, err := searchSegments(ctx, manager, segments, SegmentTypeSealed, searchReq)
	
	// 返回搜索结果、搜索的段列表和可能的错误
	return searchResults, segments, err
}

// searchStreaming will search all the target segments in streaming
// if partIDs is empty, it means all the partitions of the loaded collection or all the partitions loaded.
func SearchStreaming(ctx context.Context, manager *Manager, searchReq *SearchRequest, collID int64, partIDs []int64, segIDs []int64) ([]*SearchResult, []Segment, error) {
	if ctx.Err() != nil {
		return nil, nil, ctx.Err()
	}

	segments, err := validateOnStream(ctx, manager, collID, partIDs, segIDs)
	if err != nil {
		return nil, nil, err
	}
	searchResults, err := searchSegments(ctx, manager, segments, SegmentTypeGrowing, searchReq)
	return searchResults, segments, err
}

func SearchHistoricalStreamly(ctx context.Context, manager *Manager, searchReq *SearchRequest,
	collID int64, partIDs []int64, segIDs []int64, streamReduce func(result *SearchResult) error,
) ([]Segment, error) {
	if ctx.Err() != nil {
		return nil, ctx.Err()
	}

	segments, err := validateOnHistorical(ctx, manager, collID, partIDs, segIDs)
	if err != nil {
		return segments, err
	}
	err = searchSegmentsStreamly(ctx, manager, segments, searchReq, streamReduce)
	if err != nil {
		return segments, err
	}
	return segments, nil
}
